package com.zhang.first.day08;

import com.zhang.first.utils.ClickEvent;
import com.zhang.first.utils.ClickEventSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

/**
 * @title: Flink触发器
 * @author: zhang
 * @date: 2022/1/22 19:50
 */
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickEventSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<ClickEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                                    @Override
                                    public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                )
                .keyBy(r -> 1)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .trigger(new MyTrigger())
                .aggregate(new CountAgg(), new WindowResult())
                .print();

        env.execute();
    }

    // 窗口中的第一条数据的时间戳后面的每个整数秒，都要触发一次窗口的计算
    // 例如时间戳是1234ms的话，那么需要在2,3,4,5,6,7,8,9都要触发窗口的计算，但是不销毁窗口
    private static class MyTrigger extends Trigger<ClickEvent, TimeWindow> {
        //每来一条数据，触发一次执行
        @Override
        public TriggerResult onElement(ClickEvent element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
            ValueState<Boolean> isFirstEvent = ctx.getPartitionedState(
                    new ValueStateDescriptor<Boolean>(
                            "first",
                            Types.BOOLEAN
                    )
            );
            if (isFirstEvent.value() == null) {
                //第一条数据之后的整数秒
                // 窗口中的第一条数据到达
                // nextTime是element.ts后面的第一个整数秒
                long nextTime = element.ts + 1000L - element.ts % 1000L;
                isFirstEvent.update(true);
                ctx.registerEventTimeTimer(nextTime);
            }
            return TriggerResult.CONTINUE;
        }

        @Override
        public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            return null;
        }

        //时间时间定时器
        //当水位线到达参数 time 时候触发 onEventTime
        @Override
        public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
            if (time<window.getEnd()){
                if (time+1000L< window.getEnd()){
                    //注册还是onEventTime定时器
                    ctx.registerEventTimeTimer(time+1000L);
                }
                //触发窗口计算（也就是后面的 .aggregate方法的执行），但是不销毁窗口
                return TriggerResult.FIRE;
            }
            return TriggerResult.CONTINUE;
        }

        // 窗口闭合销毁的时候调用
        @Override
        public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
            ValueState<Boolean> isFirstEvent = ctx.getPartitionedState(
                    new ValueStateDescriptor<Boolean>(
                            "first",
                            Types.BOOLEAN
                    )
            );
            isFirstEvent.clear();
        }
    }

    public static class CountAgg implements AggregateFunction<ClickEvent, Long, Long> {
        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(ClickEvent value, Long accumulator) {
            return accumulator + 1L;
        }

        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return null;
        }
    }


    private static class WindowResult extends ProcessWindowFunction<Long, String, Integer, TimeWindow> {
        @Override
        public void process(Integer integer, ProcessWindowFunction<Long, String, Integer, TimeWindow>.Context context, Iterable<Long> elements, Collector<String> out) throws Exception {
            out.collect("窗口:" + new Timestamp(context.window().getStart()) + "~~" + new Timestamp(context.window().getEnd())
                    + "\t中的pv:" + elements.iterator().next());
        }
    }
}

